Dubbo源码分析11 -Dubbo 服务引入

计划阅读调试下Dubbo的源码，结合官方源码分析Dubbo，自身再分析总结

本文对应的Dubbo 服务引入

源码分析

invoker

接着上一篇，进入refreshInvokers

public class RegistryDirectory<T> extends AbstractDirectory<T> implements NotifyListener {
    private void refreshInvoker(List<URL> invokerUrls) {
      // 如果invokerUrl中仅有empty协议的url
      if (invokerUrls != null && invokerUrls.size() == 1 && invokerUrls.get(0) != null
              && Constants.EMPTY_PROTOCOL.equals(invokerUrls.get(0).getProtocol())) {
          this.forbidden = true; // Forbid to access
          this.methodInvokerMap = null; // Set the method invoker map to null
          destroyAllInvokers(); // Close all invokers
      } else {
          this.forbidden = false; // Allow to access
          Map<String, Invoker<T>> oldUrlInvokerMap = this.urlInvokerMap; // local reference
          if (invokerUrls.isEmpty() && this.cachedInvokerUrls != null) {
              invokerUrls.addAll(this.cachedInvokerUrls);
          } else {
              this.cachedInvokerUrls = new HashSet<URL>();
              this.cachedInvokerUrls.addAll(invokerUrls);//Cached invoker urls, convenient for comparison
          }
          if (invokerUrls.isEmpty()) {
              return;
          }
          // 转化为invokers对象列表
          Map<String, Invoker<T>> newUrlInvokerMap = toInvokers(invokerUrls);// Translate url list to Invoker map
          // 转化为methd对应的invoker
          Map<String, List<Invoker<T>>> newMethodInvokerMap = toMethodInvokers(newUrlInvokerMap); // Change method name to map Invoker Map
          // state change
          // If the calculation is wrong, it is not processed.
          if (newUrlInvokerMap == null || newUrlInvokerMap.size() == 0) {
              logger.error(new IllegalStateException("urls to invokers error .invokerUrls.size :" + invokerUrls.size() + ", invoker.size :0. urls :" + invokerUrls.toString()));
              return;
          }
          this.methodInvokerMap = multiGroup ? toMergeMethodInvokerMap(newMethodInvokerMap) : newMethodInvokerMap;
          this.urlInvokerMap = newUrlInvokerMap;
          try {
              // 关闭oldUrlInvokerMap中不在newUrlInvokerMap存在的
              destroyUnusedInvokers(oldUrlInvokerMap, newUrlInvokerMap); // Close the unused Invoker
          } catch (Exception e) {
              logger.warn("destroyUnusedInvokers error. ", e);
          }
      }
    }

  private Map<String, Invoker<T>> toInvokers(List<URL> urls) {
      Map<String, Invoker<T>> newUrlInvokerMap = new HashMap<String, Invoker<T>>();
      if (urls == null || urls.isEmpty()) {
          return newUrlInvokerMap;
      }
      Set<String> keys = new HashSet<String>();
      // 从url中获取protocol参数
      String queryProtocols = this.queryMap.get(Constants.PROTOCOL_KEY);
      for (URL providerUrl : urls) {
          // If protocol is configured at the reference side, only the matching protocol is selected
          // 如注释所说，如果配置了protocol，只接受配置允许的
          if (queryProtocols != null && queryProtocols.length() > 0) {
              boolean accept = false;
              String[] acceptProtocols = queryProtocols.split(",");
              for (String acceptProtocol : acceptProtocols) {
                  if (providerUrl.getProtocol().equals(acceptProtocol)) {
                      accept = true;
                      break;
                  }
              }
              if (!accept) {
                  continue;
              }
          }
          // 跳过空协议
          if (Constants.EMPTY_PROTOCOL.equals(providerUrl.getProtocol())) {
              continue;
          }
          // 协议检查
          if (!ExtensionLoader.getExtensionLoader(Protocol.class).hasExtension(providerUrl.getProtocol())) {
              logger.error(new IllegalStateException("Unsupported protocol " + providerUrl.getProtocol() + " in notified url: " + providerUrl + " from registry " + getUrl().getAddress() + " to consumer " + NetUtils.getLocalHost()
                      + ", supported protocol: " + ExtensionLoader.getExtensionLoader(Protocol.class).getSupportedExtensions()));
              continue;
          }
          URL url = mergeUrl(providerUrl);

          String key = url.toFullString(); // The parameter urls are sorted
          if (keys.contains(key)) { // Repeated url
              continue;
          }
          keys.add(key);
          // Cache key is url that does not merge with consumer side parameters, regardless of how the consumer combines parameters, if the server url changes, then refer again
          Map<String, Invoker<T>> localUrlInvokerMap = this.urlInvokerMap; // local reference
          Invoker<T> invoker = localUrlInvokerMap == null ? null : localUrlInvokerMap.get(key);
          if (invoker == null) { // Not in the cache, refer again
              try {
                  boolean enabled = true;
                  if (url.hasParameter(Constants.DISABLED_KEY)) {
                      enabled = !url.getParameter(Constants.DISABLED_KEY, false);
                  } else {
                    // 默认enable true
                      enabled = url.getParameter(Constants.ENABLED_KEY, true);
                  }
                  if (enabled) {
                      // protocol 在这里对应DubboProtocol，生成invoker代理
                      invoker = new InvokerDelegate<T>(protocol.refer(serviceType, url), url, providerUrl);
                  }
              } catch (Throwable t) {
                  logger.error("Failed to refer invoker for interface:" + serviceType + ",url:(" + url + ")" + t.getMessage(), t);
              }
              if (invoker != null) { // Put new invoker in cache
                  newUrlInvokerMap.put(key, invoker);
              }
          } else {
              newUrlInvokerMap.put(key, invoker);
          }
      }
      keys.clear();
      return newUrlInvokerMap;
  }

}

refreshInvoke方法的功能对当前的invoker更新，新建新的provider对应的invoker，并移除不可用的invoker

在toInvokers方法中，protocol#ref 方法中调用了各自protocol方法的逻辑。现在进入到默认的协议，DubboProtocol中

public class DubboProtocol extends AbstractProtocol {

    @Override
    public <T> Invoker<T> refer(Class<T> serviceType, URL url) throws RpcException {
        
        optimizeSerialization(url);
        // create rpc invoker.
        DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);
        invokers.add(invoker);
        return invoker;
    }

       private ExchangeClient[] getClients(URL url) {
        // 获取客户端
        // whether to share connection
        boolean service_share_connect = false;
        // 共享客户端
        int connections = url.getParameter(Constants.CONNECTIONS_KEY, 0);
        // if not configured, connection is shared, otherwise, one connection for one service
        if (connections == 0) {
            service_share_connect = true;
            connections = 1;
        }

        ExchangeClient[] clients = new ExchangeClient[connections];
        for (int i = 0; i < clients.length; i++) {
            if (service_share_connect) {
                // 获取共享客户端
                clients[i] = getSharedClient(url);
            } else {
                // 初始化客户端
                clients[i] = initClient(url);
            }
        }
        return clients;
    }

    /**
     * Get shared connection
     */
    private ExchangeClient getSharedClient(URL url) {
        String key = url.getAddress();
        // 从缓存中取
        ReferenceCountExchangeClient client = referenceClientMap.get(key);
        if (client != null) {
            if (!client.isClosed()) {
                client.incrementAndGetCount();
                return client;
            } else {
                referenceClientMap.remove(key);
            }
        }

        // 加锁处理，单个线程一个一个处理
        locks.putIfAbsent(key, new Object());
        synchronized (locks.get(key)) {
            // 从缓存中获取
            if (referenceClientMap.containsKey(key)) {
                return referenceClientMap.get(key);
            }
            // 缓存拿不到，初始化客户端
            ExchangeClient exchangeClient = initClient(url);
            client = new ReferenceCountExchangeClient(exchangeClient, ghostClientMap);
            referenceClientMap.put(key, client);
            ghostClientMap.remove(key);
            locks.remove(key);
            return client;
        }
    }

    /**
     * Create new connection
     */
    private ExchangeClient initClient(URL url) {

        // client type setting.
        // 默认客户端类型为netty
        String str = url.getParameter(Constants.CLIENT_KEY, url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_CLIENT));

        // 编码
        url = url.addParameter(Constants.CODEC_KEY, DubboCodec.NAME);
        // enable heartbeat by default
        // 心跳
        url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT));

        // BIO is not allowed since it has severe performance issue.
        if (str != null && str.length() > 0 && !ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str)) {
            throw new RpcException("Unsupported client type: " + str + "," +
                    " supported client type is " + StringUtils.join(ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions(), " "));
        }

        ExchangeClient client;
        try {
            // connection should be lazy
            if (url.getParameter(Constants.LAZY_CONNECT_KEY, false)) {
                client = new LazyConnectExchangeClient(url, requestHandler);
            } else {
                // 客户端初始化
                client = Exchangers.connect(url, requestHandler);
            }
        } catch (RemotingException e) {
            throw new RpcException("Fail to create remoting client for service(" + url + "): " + e.getMessage(), e);
        }
        return client;
    }
}

public class Exchangers {

    public static ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
        if (url == null) {
            throw new IllegalArgumentException("url == null");
        }
        if (handler == null) {
            throw new IllegalArgumentException("handler == null");
        }
        url = url.addParameterIfAbsent(Constants.CODEC_KEY, "exchange");
        return getExchanger(url).connect(url, handler);
    }

    public static Exchanger getExchanger(URL url) {
      // 默认 HeaderExchanger
        String type = url.getParameter(Constants.EXCHANGER_KEY, Constants.DEFAULT_EXCHANGER);
        return getExchanger(type);
    }
}

public class HeaderExchanger implements Exchanger {

    public static final String NAME = "header";

    @Override
    public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
      // 从数据交换层到传输曾
        return new HeaderExchangeClient(Transporters.connect(url, new DecodeHandler(new HeaderExchangeHandler(handler))), true);
    }
}
public class Transporters {

    public static Client connect(URL url, ChannelHandler... handlers) throws RemotingException {
        if (url == null) {
            throw new IllegalArgumentException("url == null");
        }
        ChannelHandler handler;
        if (handlers == null || handlers.length == 0) {
          // 没有handler则新建适配器
            handler = new ChannelHandlerAdapter();
        } else if (handlers.length == 1) {
            handler = handlers[0];
        } else {
          // 多个则创建分发器
            handler = new ChannelHandlerDispatcher(handlers);
        }
        // 默认nettyTransporter
        return getTransporter().connect(url, handler);
    }

    public static Transporter getTransporter() {
        return ExtensionLoader.getExtensionLoader(Transporter.class).getAdaptiveExtension();
    }
}

// 进入到传输层

public class NettyTransporter implements Transporter {

    @Override
    public Client connect(URL url, ChannelHandler listener) throws RemotingException {
      // 创建netty类型客户端 ， 后续涉及handler包装，消息分发等，后面再分析
        return new NettyClient(url, listener);
    }

}

后续将继续RegistryDirectory余下代码